package vip.gnloypp.redis.mq.api.core.manager;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.thread.ThreadUtil;
import com.alibaba.fastjson.JSON;
import jakarta.annotation.PostConstruct;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import vip.gnloypp.redis.mq.api.config.RedisSenderConfig;
import vip.gnloypp.redis.mq.api.config.RedisSenderQueueConfig;
import vip.gnloypp.redis.mq.api.core.base.LogManager;
import vip.gnloypp.redis.mq.api.data.RedisMqConstant;
import vip.gnloypp.redis.mq.api.data.RedisMqMessage;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author GuoNanLin
 * @since 2024/8/30 16:49
 */
public class RedisSenderManager {

    private final RedisSenderConfig redisSenderConfig;
    private final StringRedisTemplate stringRedisTemplate;
    private final LogManager logManager;

    public RedisSenderManager(RedisSenderConfig redisSenderConfig) {
        this.redisSenderConfig = redisSenderConfig;
        this.stringRedisTemplate = new StringRedisTemplate(redisSenderConfig.getRedisConnectionFactory());
        this.logManager = new LogManager(redisSenderConfig.getLogPrefix());
    }

    @PostConstruct
    private void init() {
        List<RedisSenderQueueConfig> configs = redisSenderConfig.getRedisSenderQueueConfigs();
        if (CollectionUtil.isEmpty(configs)) {
            logManager.error("未设置发送队列配置");
            return;
        }
        configs.forEach(config -> {
            logManager.notBlank(config.getQueue(), "queue不能为空");
            logManager.notNull(config.getQueueMaxSize(), "queueMaxSize不能为空");
            String key = redisSenderConfig.getEnv() + ":mq:" + config.getQueue();
            initQueue(key);
            Executors.newSingleThreadExecutor().execute(() -> {
                Long queueMinSize = (long) (config.getQueueMaxSize() * RedisMqConstant.DELETE_COEFFICIENT);
                while (true) {
                    try {
                        deleteHistoryMsg(key, config.getQueueMaxSize(), queueMinSize);
                    } catch (Exception e) {
                        logManager.error("删除历史消息，失败，key = {}", key, e);
                    } finally {
                        ThreadUtil.sleep(RedisMqConstant.DELETE_HISTORY_MSG_TIMEOUT, TimeUnit.SECONDS);
                    }
                }
            });
        });
    }

    private void initQueue(String key) {
        logManager.info("队列，初始化成功，key = {}", key);
    }

    /**
     * 删除历史消息
     *
     * @param key          队列名
     * @param queueMaxSize 队列最长长度
     * @param queueMinSize 队列最短长度
     */
    private void deleteHistoryMsg(String key, Long queueMaxSize, Long queueMinSize) {
        Long size = stringRedisTemplate.opsForStream().size(key);
        if (Objects.nonNull(size) && size >= queueMaxSize) {
            Long deleteCount = stringRedisTemplate.opsForStream().trim(key, queueMinSize);
            logManager.info("删除历史消息，key = {}，deleteCount = {}，nowCount = {}",
                    key, deleteCount, queueMinSize);
        }
    }

    public <T extends RedisMqMessage> void sendMessage(T message) {
        List<RedisSenderQueueConfig> redisSenderQueueConfigs = redisSenderConfig.getRedisSenderQueueConfigs();
        for (RedisSenderQueueConfig redisSenderQueueConfig : redisSenderQueueConfigs) {
            if (message.getClass().equals(redisSenderQueueConfig.getEventType())) {
                String queue = redisSenderQueueConfig.getQueue();
                logManager.notBlank(queue, "队列不能为空");
                String key = redisSenderConfig.getEnv() + ":mq:" + queue;
                message.setKey(key);
                RecordId recordId = stringRedisTemplate.opsForStream().add(ObjectRecord.create(key, message));
                logManager.notNull(recordId, "发送消息失败，msg = {}", JSON.toJSONString(message));
                message.setMessageId(recordId.getValue());
                message.setMessageDequeueCount(0L);
                logManager.info("发送消息，msg = {}", JSON.toJSONString(message));
                return;
            }
        }
        throw new RuntimeException("消息未配置");
    }

}
